# Databricks notebook source
# MAGIC %md
# MAGIC ## TMR Populate Task Bridge
# MAGIC <p><b>Description: </b>Populate junction tables for Task Management Report (TMR).</p>
# MAGIC <b>Parent Process: </b>
# MAGIC <table class="table table-striped table-hover">
# MAGIC  <thead>
# MAGIC   <tr>
# MAGIC    <th>Contributor</th>
# MAGIC    <th>Date</th>
# MAGIC    <th>Version</th>
# MAGIC    <th>Comment</th>
# MAGIC    <th>WorkItem No</th>
# MAGIC   </tr>
# MAGIC  </thead>
# MAGIC  <tbody>
# MAGIC   <tr>
# MAGIC    <td>Iwan Dyke</td>
# MAGIC    <td>18/04/2024</td>
# MAGIC    <td>1.0</td>
# MAGIC    <td>Create initial release</td>
# MAGIC    <td></td>
# MAGIC   </tr>
# MAGIC  </tbody>
# MAGIC </table>

# COMMAND ----------

# MAGIC %md
# MAGIC ## Import Python Libraries

# COMMAND ----------

from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.sql.window import Window
import pandas as pd  
import numpy as np 
import os  # Joining filenames.
import datetime as dt  # Adding today's date to exports.
from datetime import datetime
import glob

# COMMAND ----------

# MAGIC %md
# MAGIC ## Populate Task Owner Bridge

# COMMAND ----------

# MAGIC %sql
# MAGIC -- Clear down bridge table
# MAGIC TRUNCATE TABLE databricks_poc.tmr.dim_task_owner_bridge;
# MAGIC
# MAGIC -- Collect data for preprocessing
# MAGIC CREATE OR REPLACE TEMPORARY VIEW Bridge AS
# MAGIC SELECT DISTINCT
# MAGIC     `Task Instance ID` AS `TaskInstanceID`,
# MAGIC     CAST(NULL AS INT) AS `Code`,
# MAGIC     LTRIM(`Value`) AS `Value`,
# MAGIC     ROW_NUMBER() OVER (PARTITION BY `Task Instance ID` ORDER BY `Task Owner`) AS `Ordering`
# MAGIC FROM 
# MAGIC     databricks_poc.staging.tmr_raw f
# MAGIC     LATERAL VIEW explode(split(`Task Owner`, ',')) split AS `Value`
# MAGIC WHERE 
# MAGIC     LTRIM(`Value`) NOT LIKE 'test%'
# MAGIC     AND LTRIM(`Value`) NOT LIKE '%...%';
# MAGIC
# MAGIC -- Join back to the dimension in order to collect the code
# MAGIC CREATE OR REPLACE TEMPORARY VIEW BridgeWithCode AS
# MAGIC SELECT
# MAGIC     b.`TaskInstanceID` AS TaskInstanceID,
# MAGIC     dim.Code AS Code,
# MAGIC     b.Value,
# MAGIC     b.Ordering
# MAGIC FROM
# MAGIC     Bridge b
# MAGIC     JOIN databricks_poc.tmr.dim_task_owner dim
# MAGIC     ON dim.Description = b.Value;
# MAGIC
# MAGIC -- INSERT into the junction table
# MAGIC INSERT INTO databricks_poc.tmr.dim_task_owner_bridge
# MAGIC SELECT DISTINCT
# MAGIC   `TaskInstanceID`
# MAGIC   ,`Code`
# MAGIC   ,`Ordering`
# MAGIC FROM 
# MAGIC   BridgeWithCode;
